package com.github.distributionmessage.thread;

import com.github.distributionmessage.config.IntegrationConfiguration;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;

import java.util.concurrent.ExecutorService;

@Slf4j
public class KafkaSendMessageThread implements Runnable{

    private static ExecutorService executorService;

    private Object message;

    private KafkaTemplate kafkaTemplate;

    private String topic;

    public KafkaSendMessageThread(KafkaTemplate kafkaTemplate, Object message, String topic) {
        this.kafkaTemplate = kafkaTemplate;
        this.message = message;
        this.topic = topic;
    }

    @Override
    public void run() {
        long startTime = System.nanoTime();
        IntegrationConfiguration.CACHE_QUEUE.poll();
        this.kafkaTemplate.send(topic, message);
        log.info("cache size [" + IntegrationConfiguration.CACHE_QUEUE.size() + "] send message to topic[" + this.topic + "] use["
            + ((double)(System.nanoTime() - startTime) / 1000000.0) + "]ms");
    }

    public static ExecutorService getExecutorService() {
        return executorService;
    }

    public static void setExecutorService(ExecutorService executorService) {
        KafkaSendMessageThread.executorService = executorService;
    }
}
